/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package com.cloud.control.service.socket.handler;

import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;

import com.cloud.control.service.core.ProxyManageServices;
import com.cloud.control.service.core.RegisterUser;
import com.cloud.control.service.socket.entity.ProxyEntity;
import com.cloud.control.service.socket.entity.SenderEntity;
import com.cloud.control.service.utils.JDBCUtils;
import com.cloud.control.service.utils.RSAUtils;


/**
 * 
 * 1、这个实例对server来说是唯一的,所以必须注意Thread-Safe问题,实例变量必须线程安全!
 * 
 * 
 * 2、2015-03-05 保存一个sender和proxy实体的mapping？
 */
public class ProxyProtocolHandler extends IoHandlerAdapter {
    private final static Log log = LogFactory.getLog(ProxyProtocolHandler.class);
    private final IdleStatus interestedIdleStatus=IdleStatus.BOTH_IDLE;
    private final ProxyManageServices service=ProxyManageServices.getInstance();
    private final ExecutorService fixedThreadPool = Executors.newFixedThreadPool(6);  
    
    /**全部TCP客户端的信息，包括proxy和sender！**/
    private final ConcurrentHashMap<String,IoSession> allTCPClients=new ConcurrentHashMap<String,IoSession>();

    public void sessionCreated(IoSession session) throws Exception {
		String remoteIPAddress = ((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress();
    	session.setAttribute("state", "created");
		allTCPClients.put(remoteIPAddress, session);			
        log.info("sessionCreated:  "+remoteIPAddress);
   }
/**
 * 最常見的異常是客戶端Peer close socket!需要客戶端自己發起新連接！
 */
    @Override
    public void exceptionCaught(IoSession session, Throwable cause) {

		//cause.printStackTrace();
		try {
			String remoteIPAddress = ((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress();
			session.setAttribute("state", "exception");
			log.info("TCP connecion exception with IP == "+remoteIPAddress);
			String userid=(String) session.getAttribute("userid");
			allTCPClients.remove(remoteIPAddress);

			log.info("session Exception with IP == "+remoteIPAddress);
			//log.error(cause.getMessage());
			session.close(true);
			session=null;
		} catch (Exception e) {
			log.error("exceptionCaught exception. "+e.getCause());
		}
    }
    
    public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
    	try {
    		String remoteIPAddress = ((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress();
        	session.setAttribute("state", "idle");
    		log.info("sessionIdle exception: "+ remoteIPAddress);
 			allTCPClients.remove(remoteIPAddress);
			
			session.close(true);
		} catch (Exception e) {
			log.error("sessionIdle exception. "+e.getCause());
		}
    }

    @Override
    //记录Client的TCP session连接开始的情况!
    public void sessionOpened(IoSession session) throws Exception {
    	session.getConfig().setIdleTime(interestedIdleStatus, 30);//30秒idle
    	session.setAttribute("state", "opened");
		String remoteIPAddress = ((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress();
		allTCPClients.put(remoteIPAddress, session);			
        log.info("session Opened. with remote IP address:=  "+ remoteIPAddress);
    }

    @Override
    public void messageReceived(final IoSession session, Object message) {
       
    	String remoteIPAddress=((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress();
    	
	    /************处理 proxy的协议****************/
        if (message instanceof ProxyEntity)
        {
        	Object proxySession= session.getAttribute("proxy");
        	ProxyEntity proxy = (ProxyEntity) message;
            if(proxy!=null){
            	proxy.setIpAddress(remoteIPAddress);
            }
            
	        try {
	        	//回应heat beat
	        	String commandIn=proxy.getCommandIn();
	        	Object cert=session.getAttribute("cert");
	        	if("failed".equalsIgnoreCase((String)cert)){
           			session.write("proxy----REFUSED#");
           			session.close(true);
           			return;
        		}
	        	
	        	if(commandIn.equals("heartbeat")){
	        		//获取proxy对象,包含了端口的信息
	        		if(proxySession!=null){
	        			ProxyEntity temp=(ProxyEntity)proxySession;
	        			proxy.setAllocateTime(temp.getAllocateTime());
	        		}else{
	        			session.setAttribute("proxy",proxy);
	        		}
	        		if(cert==null){
	        			session.write("proxy----identify#");
	        			return;
	        		} 
	        		//验证完毕的代理才会加入队列!
	        		if("verified".equalsIgnoreCase((String)cert)){
	        			session.write("proxy----OK#");
	        		}
	        	//代理proxy初始化时从服务器请求证书！通过线程来执行！以免延误执行！
	        	}else if(commandIn.equals("register")){
	        		final String userid=proxy.getUserid();
					
	        		final IoSession sessionFin=session;
	        		RegisterUser reg=new RegisterUser();
	        		reg.setSessionFin(sessionFin);
	        		reg.setUserid(userid);
	        		reg.setName(proxy.getName());
	        		fixedThreadPool.execute(reg);//启动注册线程!
	        		
	        	//身份验证,解密耗时，需要考虑异步处理？
	        	}else if(commandIn.equals("identify")){
	        		proxy=service.validateUser(proxy);//app加密的证书
	        		if(proxy.getUserid()!=null){
	        			session.setAttribute("cert","verified");//表示session通过验证
	        			session.setAttribute("userid",proxy.getUserid());//表示session通过验证
	        			//把session和proxy对象绑定起来！
		        		if(proxySession!=null){
		        			ProxyEntity temp=(ProxyEntity)proxySession;
		        			proxy.setAllocateTime(temp.getAllocateTime());
		        		}else{
		        			session.setAttribute("proxy",proxy);
		        		}
	        			service.proxyIn(proxy);
	        			session.write("proxy----OK#");
			    		log.error("proxy "+proxy.getUserid()+" registered  "+proxy.getIpAddress()+":"+proxy.getPort());
	        		}else{
	        			session.write("proxy----FAILED#");
	        			session.setAttribute("cert","failed");  //表示session通过验证
	        		}
	        	}
	
	        } catch (IllegalArgumentException e) {
	        	log.info("Illegal argument", e.getCause());
	        }

	        
	  /************处理sender协议****************/
        }else if (message instanceof SenderEntity){
        	final SenderEntity sender = (SenderEntity) message;
            if(sender!=null){
            	sender.setTaskIP(remoteIPAddress);
            }
	        try {
	        	//回应heat beat
	        	Object cert=session.getAttribute("cert");
	        	String commandIn=sender.getCommandIn();
	        	if(cert==null){
	        		session.write("sender----identify#");
	        	}else if(commandIn.equals("heartbeat") 
	        			&& "verified".equalsIgnoreCase((String)cert)){
	        		session.write("sender----OK#");
	        	}else if("failed".equalsIgnoreCase((String)cert)){
	        		session.write("sender----REFUSED#");//拒绝连接
	        		session.close(true);
	        	}
	        	
	        	//身份验证,通过的加入管理queue
	        	if(commandIn.equals("identify")){
	        		SenderEntity sender1=service.validateUser(sender);
	        		if(sender1.getUserid()!=null){
	        			session.setAttribute("cert","verified");//表示session通过验证
	        			session.setAttribute("userid",sender1.getUserid());//表示session通过验证
	        			service.allSender.add(sender1);
	        			session.write("sender----OK#");
	        		}else{										//session校验失败
	        			session.write("sender----FAILED#");
	        			session.setAttribute("cert","failed");  //表示session通过验证
	        		}
	        	//sender发起注册请求,服务器需要从数据库中返回对应的证书!
	        	}else if(commandIn.equals("register")){
	        		final String userid=sender.getUserid();
					
	        		final IoSession sessionFin=session;
	        		RegisterUser reg=new RegisterUser();
	        		reg.setSessionFin(sessionFin);
	        		reg.setUserid(userid);
	        		//启动注册线程!
	        		fixedThreadPool.execute(reg);
	        		
	        	//分配代理proxy给sender
	        	}else if(commandIn.equals("proxies")){
	        		String flag=(String)session.getAttribute("cert");
	        		if(flag.equals("verified")){
	        			fixedThreadPool.execute(
	        				new Runnable(){
								@Override
								public void run() {
					        		String trustIPs=service.allocateProxies4Task(sender);//把ip回写给proxy了
					        		sender.setTrustIPs(trustIPs);
					        		log.error("allocate proxies to sender: "+trustIPs);
				        			session.write("sender----ipaddress----"+trustIPs+"#");
								}
	        					
	        				}
	        			);
	        		}else{										
	        			session.write("sender----FAILED#");
	        			session.setAttribute("cert","failed");  
	        		}
	        	}else if(commandIn.equals("pCount")){//获取在线代理的数量！
	        		String flag=(String)session.getAttribute("cert");
	        		if(flag.equals("verified")){
	        			int pCount=service.proxyQueue.size();
	        			session.write("sender----pCount----"+pCount+"#");
	        		}else{										
	        			session.write("sender----FAILED#");
	        			session.setAttribute("cert","failed");  
	        		}
	        	}
		    	/**增加遠程刪除proxy的能力！**/
	        	else if(commandIn.equals("remove")){//获取在线代理的数量！
	        		String removedIP= sender.getTaskType();
	        		IoSession session1=allTCPClients.get(removedIP);
	        		if(removedIP!=null && session1!=null){
	        			session1.close(true);
		        		String userid=(String) session.getAttribute("userid");
		        		service.proxyOut(userid);
		        		log.error(" remote proxy removed from queue" +userid+"--"+removedIP);
	        		}
	        	}
	
	        } catch (IllegalArgumentException e) {
	        	log.info("Illegal argument", e.getCause());
	        }
        }
    }



    @Override
    public void sessionClosed(IoSession session) throws Exception {
    	try {
    		String remoteIPAddress=((InetSocketAddress)session.getRemoteAddress()).getAddress().getHostAddress();
        	session.setAttribute("state", "closed");
        	String userid=(String)session.getAttribute("userid");
    		log.error("session closed: "+ remoteIPAddress+" - userid: "+userid);
			allTCPClients.remove(remoteIPAddress);

			/*不能判断session的属性，就同时从proxy和sender队列中remove！*/
			if(service!=null && userid!=null){
				service.proxyOut(userid);
			}
			service.allSender.remove(remoteIPAddress);
			
			session.close(true);
			
	        session=null;
		} catch (Exception e) {
			log.debug("session closed exception. "+e.getCause());
		}
    }


}
